package com.duoduo.demo.rabbitmq.rpc;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.log4j.Logger;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Queue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

/**
 * 基于rabbitMQ的RPC服务的服务调用者 消费生产者角色，发送调用请求，然后读取回复 发送的交换器 和 接收回应的交换器是不同的
 * @author: yzl
 * @date: 2016-10-23
 */
public class RpcClient {

	// 交换器名称
	private static final String EXCHANGE_NAME = "demo.rpc.ex";
	// 服务提供方的路由key
	private static final String SERVER_ROUTING_KEY = "demo.rpc.k";

	private static final Logger logger = Logger.getLogger(RpcClient.class);

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("172.16.185.233");
		factory.setPort(5672);
		factory.setVirtualHost("/");
		factory.setUsername("dev");
		factory.setPassword("dev");

		Connection connection = factory.newConnection();
		final Channel channel = connection.createChannel();

		channel.exchangeDeclare(EXCHANGE_NAME, "direct");

		int threadCount = 1;

		// 使用CountDownLatch控制10个线程一起运行
		final CountDownLatch cdl = new CountDownLatch(threadCount);

		// 生成10个线程同时访问服务
		ExecutorService pool = Executors.newFixedThreadPool(threadCount);

		for (int i = 0; i < threadCount; i++) {
			final int index = i;
			pool.submit(new Runnable() {

				@Override
				public void run() {
					try {
						cdl.await();

						// Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean
						// autoDelete,Map<String, Object> arguments)
						// 默认direct exchange 匿名的、 私有、自动删除
						Queue.DeclareOk queue = channel.queueDeclare("", false, true, true, null);
						// 获取rabbit帮我们生成的随即的队列名称，来接收服务端返回的数据
						String queueName = queue.getQueue();

						final String messageId = UUID.randomUUID().toString();
						// 定义replyTo属性 和 消息ID
						BasicProperties props = new BasicProperties.Builder().replyTo(queueName)
								.correlationId(messageId).build();

						logger.info("发送rpc调用请求，消息index:" + index);

						// 发送RPC方法调用
						channel.basicPublish(EXCHANGE_NAME, SERVER_ROUTING_KEY, props,
								("add(" + index + "," + (index + 1) + ")").getBytes());

						logger.info("等待服务器响应");

						// 定义队列式的消费者处理器，之前是用的DefaultConsumer
						QueueingConsumer consumer = new QueueingConsumer(channel);

						// 把消费者处理器和队列对照起来
						channel.basicConsume(queueName, consumer);

						// 这里会堵塞，直到取到值或者超时
						logger.info("尝试从consumer里取返回的值");
						Delivery delivery = consumer.nextDelivery();
						// Delivery delivery = consumer.nextDelivery(5000);
						logger.info("成功取到消息，开始处理");
						if (delivery.getProperties().getCorrelationId().equals(messageId)) {
							logger.info("收到服务器回复-----------");
							String msg = new String(delivery.getBody());
							logger.info("回复消息id：" + delivery.getProperties().getCorrelationId() + "内容:" + msg);
						}

						/*
						 * //该取消息的方式是异步的，不会堵塞，会导致后面的logger.info先于handleDelivery执行 channel.basicConsume(queueName, new
						 * DefaultConsumer(channel){
						 * @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties
						 * properties, byte[] body) throws IOException {
						 * if(properties.getCorrelationId().equals(messageId)){ logger.info("收到服务器回复-----------");
						 * String msg = new String(body); logger.info("回复消息id：" + properties.getCorrelationId() + "内容:"
						 * + msg); } } });
						 */

						logger.info("消息处理完成");
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			});
		}
		for (int i = 0; i < threadCount; i++) {
			cdl.countDown();
		}
	}
}
